```python
import asyncio
import os
import shlex
from typing import Optional

from app.tool.base import BaseTool, CLIResult


class Terminal(BaseTool):
    """
    A class to execute CLI commands on the system asynchronously.

    Attributes:
        name (str): The name of the tool.
        description (str): A description of the tool's purpose.
        parameters (dict): The parameters required to execute the command.
        process (Optional[asyncio.subprocess.Process]): The current running process.
        current_path (str): The current working directory.
        lock (asyncio.Lock): A lock to ensure thread-safe execution.
    """

    name: str = "execute_command"
    description: str = """请求在系统上执行CLI命令。
当你需要执行系统操作或运行特定命令来完成用户的任务时使用此命令。
你必须根据用户的系统定制你的命令，并提供清晰的命令功能解释。
优先执行复杂的CLI命令而不是创建可执行脚本，因为它们更灵活且更易于运行。
命令将在当前工作目录中执行。
注意：对于将在50ms内完成的命令，必须在命令末尾附加`sleep 0.05`，以规避终端工具中已知的命令完成过快时无法返回输出的问题。"""
    parameters: dict = {
        "type": "object",
        "properties": {
            "command": {
                "type": "string",
                "description": "(必需) 要执行的CLI命令。该命令应适用于当前操作系统。确保命令格式正确且不包含任何有害指令。",
            }
        },
        "required": ["command"],
    }
    process: Optional[asyncio.subprocess.Process] = None
    current_path: str = os.getcwd()
    lock: asyncio.Lock = asyncio.Lock()

    async def execute(self, command: str) -> CLIResult:
        """
        异步执行带有持久上下文的终端命令。

        参数:
            command (str): 要执行的终端命令。

        返回:
            CLIResult: 命令执行的输出和错误。
        """
        # 按&分割命令以处理多个命令
        commands = [cmd.strip() for cmd in command.split("&") if cmd.strip()]
        final_output = CLIResult(output="", error="")

        for cmd in commands:
            sanitized_command = self._sanitize_command(cmd)

            # 内部处理'cd'命令
            if sanitized_command.lstrip().startswith("cd "):
                result = await self._handle_cd_command(sanitized_command)
            else:
                async with self.lock:
                    try:
                        self.process = await asyncio.create_subprocess_shell(
                            sanitized_command,
                            stdout=asyncio.subprocess.PIPE,
                            stderr=asyncio.subprocess.PIPE,
                            cwd=self.current_path,
                        )
                        stdout, stderr = await self.process.communicate()
                        result = CLIResult(
                            output=stdout.decode().strip(),
                            error=stderr.decode().strip(),
                        )
                    except Exception as e:
                        result = CLIResult(output="", error=str(e))
                    finally:
                        self.process = None

            # 合并输出
            if result.output:
                final_output.output += (
                    (result.output + "\n") if final_output.output else result.output
                )
            if result.error:
                final_output.error += (
                    (result.error + "\n") if final_output.error else result.error
                )

        # 去除末尾的换行符
        final_output.output = final_output.output.rstrip()
        final_output.error = final_output.error.rstrip()
        return final_output

    async def execute_in_env(self, env_name: str, command: str) -> CLIResult:
        """
        在指定的Conda环境中异步执行终端命令。

        参数:
            env_name (str): Conda环境的名称。
            command (str): 要在环境中执行的终端命令。

        返回:
            CLIResult: 命令执行的输出和错误。
        """
        sanitized_command = self._sanitize_command(command)

        # 构造在Conda环境中运行的命令
        # 使用'conda run -n env_name command'来执行而不激活
        conda_command = f"conda run -n {shlex.quote(env_name)} {sanitized_command}"

        return await self.execute(conda_command)

    async def _handle_cd_command(self, command: str) -> CLIResult:
        """
        处理'cd'命令以更改当前路径。

        参数:
            command (str): 要处理的'cd'命令。

        返回:
            CLIResult: 'cd'命令的结果。
        """
        try:
            parts = shlex.split(command)
            if len(parts) < 2:
                new_path = os.path.expanduser("~")
            else:
                new_path = os.path.expanduser(parts[1])

            # 处理相对路径
            if not os.path.isabs(new_path):
                new_path = os.path.join(self.current_path, new_path)

            new_path = os.path.abspath(new_path)

            if os.path.isdir(new_path):
                self.current_path = new_path
                return CLIResult(
                    output=f"Changed directory to {self.current_path}", error=""
                )
            else:
                return CLIResult(output="", error=f"No such directory: {new_path}")
        except Exception as e:
            return CLIResult(output="", error=str(e))

    @staticmethod
    def _sanitize_command(command: str) -> str:
        """
        对命令进行消毒以安全执行。

        参数:
            command (str): 要消毒的命令。

        返回:
            str: 消毒后的命令。
        """
        # 示例消毒：限制某些危险命令
        dangerous_commands = ["rm", "sudo", "shutdown", "reboot"]
        try:
            parts = shlex.split(command)
            if any(cmd in dangerous_commands for cmd in parts):
                raise ValueError("Use of dangerous commands is restricted.")
        except Exception:
            # 如果shlex.split失败，尝试基本字符串比较
            if any(cmd in command for cmd in dangerous_commands):
                raise ValueError("Use of dangerous commands is restricted.")

        # 可以在此处添加额外的消毒逻辑
        return command

    async def close(self):
        """如果存在，则关闭持久shell进程。"""
        async with self.lock:
            if self.process:
                self.process.terminate()
                try:
                    await asyncio.wait_for(self.process.wait(), timeout=5)
                except asyncio.TimeoutError:
                    self.process.kill()
                    await self.process.wait()
                finally:
                    self.process = None

    async def __aenter__(self):
        """进入异步上下文管理器。"""
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """退出异步上下文管理器并关闭进程。"""
        await self.close()
```